package com.zyh.kafka.day03;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class TransactionProducerTest {
    public static void main(String[] args) {
        //1 配置生产者参数
        Properties producerConfigs = new Properties();
        producerConfigs.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"kafka24:9092");
        producerConfigs.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        producerConfigs.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 开启事务功能
        producerConfigs.setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG,"transaction-test");
        //2 创建生产者
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(producerConfigs);
        //a 初始化事务
        kafkaProducer.initTransactions();
        //b 开启事务
        kafkaProducer.beginTransaction();
        try {
            //3 发送消息
            kafkaProducer.send(new ProducerRecord<>("topic-a","hello-a"));
            kafkaProducer.send(new ProducerRecord<>("topic-b","hello-b"));
            //c 提交事务
            kafkaProducer.commitTransaction();
        } catch (Exception e) {
            //d 回滚事务
            kafkaProducer.abortTransaction();
        } finally {
            //4 关闭生产者
            kafkaProducer.close();
        }
    }
}
